package org.apache.flink.formats.json;

import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.serialization.SerializationSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.formats.common.TimestampFormat;
import org.apache.flink.table.connector.ChangelogMode;
import org.apache.flink.table.connector.Projection;
import org.apache.flink.table.connector.format.DecodingFormat;
import org.apache.flink.table.connector.format.EncodingFormat;
import org.apache.flink.table.connector.format.ProjectableDecodingFormat;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.factories.DeserializationFormatFactory;
import org.apache.flink.table.factories.DynamicTableFactory;
import org.apache.flink.table.factories.FactoryUtil;
import org.apache.flink.table.factories.SerializationFormatFactory;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.RowType;

import java.util.Collections;
import java.util.HashSet;
import java.util.Set;

@Internal
public class CustomDeserializationFormatFactory implements DeserializationFormatFactory, SerializationFormatFactory {
    public static final String IDENTIFIER = "jusz-json";

    public CustomDeserializationFormatFactory() {
    }

    @Override
    public DecodingFormat<DeserializationSchema<RowData>> createDecodingFormat(DynamicTableFactory.Context context, ReadableConfig formatOptions) {
        FactoryUtil.validateFactoryOptions(this, formatOptions);
        JsonFormatOptionsUtil.validateDecodingFormatOptions(formatOptions);
        final boolean failOnMissingField = (Boolean)formatOptions.get(JsonFormatOptions.FAIL_ON_MISSING_FIELD);
        final boolean ignoreParseErrors = (Boolean)formatOptions.get(JsonFormatOptions.IGNORE_PARSE_ERRORS);
        final TimestampFormat timestampOption = JsonFormatOptionsUtil.getTimestampFormat(formatOptions);
        return new ProjectableDecodingFormat<DeserializationSchema<RowData>>() {
            public DeserializationSchema<RowData> createRuntimeDecoder(org.apache.flink.table.connector.source.DynamicTableSource.Context context, DataType physicalDataType, int[][] projections) {
                DataType producedDataType = Projection.of(projections).project(physicalDataType);
                RowType rowType = (RowType)producedDataType.getLogicalType();
                TypeInformation<RowData> rowDataTypeInfo = context.createTypeInformation(producedDataType);
                return new JuszJsonRowDataDeserializationSchema(rowType, rowDataTypeInfo, failOnMissingField, ignoreParseErrors, timestampOption);
            }

            public ChangelogMode getChangelogMode() {
                return ChangelogMode.insertOnly();
            }
        };
    }

    @Override
    public EncodingFormat<SerializationSchema<RowData>> createEncodingFormat(DynamicTableFactory.Context context, ReadableConfig formatOptions) {
        FactoryUtil.validateFactoryOptions(this, formatOptions);
        JsonFormatOptionsUtil.validateEncodingFormatOptions(formatOptions);
        final TimestampFormat timestampOption = JsonFormatOptionsUtil.getTimestampFormat(formatOptions);
        final JsonFormatOptions.MapNullKeyMode mapNullKeyMode = JsonFormatOptionsUtil.getMapNullKeyMode(formatOptions);
        final String mapNullKeyLiteral = (String)formatOptions.get(JsonFormatOptions.MAP_NULL_KEY_LITERAL);
        final boolean encodeDecimalAsPlainNumber = (Boolean)formatOptions.get(JsonFormatOptions.ENCODE_DECIMAL_AS_PLAIN_NUMBER);
        return new EncodingFormat<SerializationSchema<RowData>>() {
            public SerializationSchema<RowData> createRuntimeEncoder(org.apache.flink.table.connector.sink.DynamicTableSink.Context context, DataType consumedDataType) {
                RowType rowType = (RowType)consumedDataType.getLogicalType();
                return new JsonRowDataSerializationSchema(rowType, timestampOption, mapNullKeyMode, mapNullKeyLiteral, encodeDecimalAsPlainNumber);
            }

            public ChangelogMode getChangelogMode() {
                return ChangelogMode.insertOnly();
            }
        };
    }

    @Override
    public String factoryIdentifier() {
        return IDENTIFIER;
    }

    @Override
    public Set<ConfigOption<?>> requiredOptions() {
        return Collections.emptySet();
    }

    @Override
    public Set<ConfigOption<?>> optionalOptions() {
        Set<ConfigOption<?>> options = new HashSet();
        options.add(JsonFormatOptions.FAIL_ON_MISSING_FIELD);
        options.add(JsonFormatOptions.IGNORE_PARSE_ERRORS);
        options.add(JsonFormatOptions.TIMESTAMP_FORMAT);
        options.add(JsonFormatOptions.MAP_NULL_KEY_MODE);
        options.add(JsonFormatOptions.MAP_NULL_KEY_LITERAL);
        options.add(JsonFormatOptions.ENCODE_DECIMAL_AS_PLAIN_NUMBER);
        return options;
    }

    @Override
    public Set<ConfigOption<?>> forwardOptions() {
        Set<ConfigOption<?>> options = new HashSet();
        options.add(JsonFormatOptions.TIMESTAMP_FORMAT);
        options.add(JsonFormatOptions.MAP_NULL_KEY_MODE);
        options.add(JsonFormatOptions.MAP_NULL_KEY_LITERAL);
        options.add(JsonFormatOptions.ENCODE_DECIMAL_AS_PLAIN_NUMBER);
        return options;
    }
}
